#include "transContentPacketMemoryIn.hpp"

//////////////////////////////////////////////////////////////
/// just write data, even over write the record
///
///	parameters:
///		content: the received data content
///		index: the content index
///
///	return:
///		E_TRANC_SUCESS: write finished
////////////////////////////////////////////////////////////////
int writeData(char* content, int index){
	char* buffer;
	debug_enter();

	/// begin to write
	if(mAvaliableCount == mCapacity){
		logd_error("packet data in, the buffer full, begin to over write, reset values");
		mAvaliableIndex = (mAvaliableIndex + 1 )%mCapacity;
		/// get avalaible start index buffer
		buffer = TransContentMemory::getInstance()->indexToBuffer(recordBuffer[mAvaliableIndex]);
		mRecordInnerOffset = getPacketHeaderLen(buffer);
		mRecordLeftBytes = getPacketDataLength(buffer);
	}else{
		++mAvaliableCount;
	}

	logd_info("packet data in, write data to buffer");
	recordBuffer[mInsertIndex] = index;
	mInsertIndex = (mInsertIndex + 1) % %mCapacity;
	return  E_TRANC_SUCESS;
}

int receiveFrom(int flag, void* buffer, int bufferLen, int timeout, receiveFromSyncCallback rfscb){
	if(receiving == 1){
		logd_error("receivefrom, the receiver busy");
		return E_TRANC_STATE_BUSY;
	}
	if(mAvaliableCount == 0 && flag == RECEIVE_FLAG_NONBLOCK){
		logd_error("receivefrom, user set flag to no block");
		return E_TRANC_NONBLOCK;
	}	
	this->dataStoreBuffer = buffer;
	this->dataStoreBufferLen = bufferLen;

	this->timeout = timeout;
	this->rfscb = rfscb;

	/// set to get data
	newReceivedDataRequest();
	if(this->rfscb != NULL){
		logd_info("receive from, use async style to receive data");
		return E_TRANC_NONBLOCK;
	}

	pthread_mutex_lock(&waitMutex);
	if(pthread_cond_timedwait(&waitCond, &waitCond, &ts) == ETIMEDOUT){
		receiving = 0;
		logd_error("receive from , timed out");
		return E_TRANC_TIMEOUT;
	}
	pthread_mutex_unlock(&waitMutex);
	return readBytes;
}

int getData(){
	char* buffer = this->dataStoreBuffer;
	int len = this->dataStoreBufferLen;
	int __leftByte = len;
	int __readBytes = 0;
	char* dataBuffer;
	int dataIndex;

	/// check there still some data to read
	if(mAvaliableCount == 0){
		logd_error("packet memory in, get data, no more data avaliable");
		return E_TRANC_NO_DATA_AVALIABLE;
	}

	while(_leftByte > 0){
		dataBuffer = TransContentMemory::getInstance()->indexToBuffer(recordBuffer[mAvaliableIndex]);
		dataIndex = recordBuffer[mAvaliableIndex];
		if(_leftByte > mRecordLeftBytes){
			/// can store the buffer data once
			memcpy(buffer+__readBytes, dataBuffer + mRecordInnerOffset, mRecordLeftBytes);
			_leftByte -= mRecordLeftBytes;
			__readBytes += mRecordLeftBytes;
			logd_info("packet memory in, get data, get the %d index buffer finished, release the buffer");
			TransContentMemory::getInstance()->releaseBuffer(dataIndex);
			--mAvaliableCount;

			//get next avaliable data
			if(mAvaliableCount > 0){
				logd_info("packet memroy in, get data, get a avaliable record finished, there still a avaliable one, rest value");
				dataBuffer = TransContentMemory::getInstance()->indexToBuffer(recordBuffer[mAvaliableIndex]);
				mRecordInnerOffset = getPacketHeaderLen(dataBuffer);
				mRecordLeftBytes = getPacketDataLength(dataBuffer);
				continue;
			}else{
				readBytes = __readBytes;
				return __readBytes;
			}
		}else{
			memcpy(buffer+__readBytes, dataBuffer + mRecordInnerOffset, _leftByte);
			__readBytes += _leftByte;
			mRecordInnerOffset += _leftByte;
			mRecordLeftBytes -= _leftByte;
			_leftByte = 0;
			readBytes = __readBytes;
			return __readBytes;
		}
	}
	logd_info("get data finished");
	receiveFinished();
	return E_TRANC_SUCESS;
}

int response(char* content, int index){
	int mainCode = content[0];
	int subCode = content[1];
	if(mainCode != message_type_out_data_packet){
		logd_error("packet memory in, response, not support maincdoe: %d", mainCode);
		return RESPONSE_RESULT_POST_CONITNUE;
	}
	switch(subCode){
		case message_type_data_packet_sub_dataIn:
			logd_info("packet memory in, response, data in");
			writeData(content, index);
			return RESPONSE_RESULT_NOACTION;
		default:
			logd_error("packet memroy in, response, not support sub code");
			return RESPONSE_RESULT_POST_CONITNUE;
	}
	logd_error("packet memroy in, response, should not go here");
	return RESPONSE_RESULT_POST_CONITNUE;
}

void TransStreamInMemory::newReceivedDataRequest(){
	pthread_mutex_lock(&mDataAvaliableMutex);
	pthread_cond_signal(&mDataAvaliableCond);
	pthread_mutex_unlock(&mDataAvaliableMutex);
}

void* TransStreamInMemory::receiveRoute(void* p){
	(TransStreamInMemory*(p))->receiveWorker();
	return NULL;
}

void TransStreamInMemory::receiveWorker(){
	while(working){
	pthread_mutex_lock(&mDataAvaliableMutex);
	pthread_cond_wait(&mDataAvaliableCond, &mDataAvaliableMutex);
	pthread_mutex_unlock(&mDataAvaliableMutex);
	getData();
}
